package com.base.data.elasticsearch.core.listener;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.NotifyOnceListener;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkResponse;

import java.util.HashMap;
import java.util.Map;

/**
 * org.elasticsearch elasticsearch 5.6.16版本 使用这个
 */
public class AsyncNotifyOnceListener extends NotifyOnceListener<BulkResponse> {
    @Override
    protected void innerOnResponse(BulkResponse bulkResponse) {
        checkForBulkUpdateFailure(bulkResponse);
    }

    @Override
    protected void innerOnFailure(Exception e) {
        throw new ElasticsearchException(
                "Bulk indexing has failures. Use ElasticsearchException.getFailedDocuments() for detailed messages ["
                        + e + "]",e);
    }

    public static void checkForBulkUpdateFailure(BulkResponse bulkResponse) {
        if (bulkResponse.hasFailures()) {
            Map<String, String> failedDocuments = new HashMap<>();
            for (BulkItemResponse item : bulkResponse.getItems()) {
                if (item.isFailed()){
                    failedDocuments.put(item.getId(), item.getFailureMessage());
                }
            }
            throw new ElasticsearchException(
                    "Bulk indexing has failures. Use ElasticsearchException.getFailedDocuments() for detailed messages ["
                            + failedDocuments + "]",
                    failedDocuments);
        }
    }

}
